package cn.ac.iscas.service.transformer;

import cn.ac.iscas.base.utils.ParameterUtils;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/8/12 11:04
 */
public class SelectColumnsTransformer {
    private final static String ID = "select_column";
    private final static String COLUMNS_KEY = "columns";
    private final static String COLUMN_NAME_KEY = "name";
    private final static String COLUMN_TYPE_KEY = "type";
    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<FlinkSourceColumn> frontColumns;

    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public SelectColumnsTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public SelectColumnsTransformer(String jobId, String taskId, String frontTableName, List<FlinkSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
    }


    public SelectColumnsTransformer() {

    }

    //String sql = "CREATE VIEW middle_platform.db1.view2 AS (Select * , 'a' AS C100 from middle_platform.db1.iganalysis_vazu WHERE C1='1002' )";
    public String getSql(Map<String, Object> parameters) {
        this.currentTableName = String.format("%s.%s.%s_%s", jobId, jobId, ID, UUID.randomUUID().toString().substring(0,6));
        String createViewSql = String.format("CREATE VIEW %s AS ", this.currentTableName);
        List<FlinkSourceColumn> columns = ParameterUtils.getRequiredListObj(parameters, COLUMNS_KEY, FlinkSourceColumn.class);
        //清空之前的列，只留当前算子选择的列
        this.currentColumns.clear();
        this.currentColumns = columns;

        String columnsSql = (columns.stream().map(FlinkSourceColumn::getName).collect(Collectors.joining(",")));
        String selectSql = String.format("( SELECT %s FROM %s )", columnsSql, frontTableName);
        createViewSql += selectSql;

        updateSource();

        return createViewSql;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
